package com.xb.system.service.impl;

import com.xb.model.domain.Users;
import com.xb.system.constants.BatchConstants;
import com.xb.system.constants.ThreadPoolManager;
import com.xb.system.service.DynamicDataSplitterService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;


@Slf4j
@Component
public class BatchProcessor {

    public static void processQueue(DynamicDataSplitterService dynamicDataSplitterService, int batchSize) {
        BlockingQueue<Users> queue = BatchConstants.QUEUE;
        batchSize = batchSize / 10;
        List<Users> buffer = new ArrayList<>(batchSize);

        try {
            while (true) {
                Users user = queue.poll(200, TimeUnit.MILLISECONDS);
                if (user != null) {
                    buffer.add(user);
                }
                
                if (buffer.size() >= batchSize || (BatchConstants.SIGNAL_COMPLETION.get() && !buffer.isEmpty() && queue.isEmpty())) {
                    //1、如果buffer队列满了，走这一步
                    //2、如果buffer队列没满，但是主线程已经读完数据了，并且阻塞队列已经空了，并且buffer还有剩余一点数据，走这一步
                    dynamicDataSplitterService.saveBatch(buffer);

                    if (BatchConstants.SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("批处理线程被中断", e);
        }
    }
    
}

